其他
大而全面｜Logstash技术入门
Logstash简介
Logstash是一个接收，处理，转发日志的工具。支持系统日志，webserver日志，错误日志，应用日志，总之包括所有可以抛出来的日志类型。怎么样听起来挺厉害的吧？
依赖条件：JAVA
Logstash运行仅仅依赖java运行环境(jre)。各位可以在命令行下运行java -version命令
启动和运行Logstash的两条命令示例
第一步我们先下载Logstash
curl -O https://artifacts.elastic.co/downloads/logstash/logstash-6.4.0.tar.gz现在你应该有了一个叫logstash-6.4.0.tar.gz的文件了。 我们把它解压一下
tar zxvf logstash-6.4.0.tar.gz cd logstash-6.4.0现在我们来运行一下：
bin/logstash -e 'input { stdin { } } output { stdout {} }'我们现在可以在命令行下输入一些字符，然后我们将看到logstash的输出内容：
hello world
2013-11-21T01:22:14.405+0000 0.0.0.0 hello world
以上例子我们在运行logstash中，定义了一个叫"stdin"的input还有一个"stdout"的output，无论我们输入什么字符，Logstash都会按照某种格式来返回我们输入的字符。这里注意我们在命令行中使用了 -e 参数，该参数允许Logstash直接通过命令行接受设置。这点尤其快速的帮助我们反复的测试配置是否正确而不用写配置文件。
让我们再试个更有意思的例子。首先我们在命令行下使用CTRL-C命令退出之前运行的Logstash。现在我们重新运行Logstash使用下面的命令：
bin/logstash -e 'input { stdin { } } output { stdout { codec => rubydebug } }'我们再输入一些字符，这次我们输入"goodnight moon"：
goodnight moon
{
"message" => "goodnight moon",
"@timestamp" => "2013-11-20T23:48:05.335Z",
"@version" => "1",
"host" => "my-laptop"
}
以上示例通过重新设置了叫"stdout"的output(添加了"codec"参数)，我们就可以改变Logstash的输出表现。类似的我们可以通过在你的配置文件中添加或者修改inputs、outputs、filters，就可以使随意的格式化日志数据成为可能，从而订制更合理的存储格式为查询提供便利。
使用Elasticsearch存储日志
现在，你也许会说："它看起来还挺高大上的，不过手工输入字符，并把字符从控制台回显出来。实际情况并不实用"。说的好，那么接下来我们将建立Elasticsearch来存储输入到Logstash的日志数据。如果你还没有安装Elasticsearch，你可以 下载RPM/DEB包 或者手动下载tar包，通过以下命令：
curl -O https://download.elasticsearch.org/elasticsearch/elasticsearch/elasticsearch-6.4.0.tar.gz
tar zxvf elasticsearch-6.4.0.tar.gz
cd elasticsearch-6.4.0/
./bin/elasticsearch
注意
本篇文章实例使用Logstash 6.4.0和Elasticsearch 6.4.0。不同的Logstash版本都有对应的建议Elasticsearch版本。请确认你使用的Logstash版本！
更多有关安装和设置Elasticsearch的信息可以参考 Elasticsearch官网 。因为我们主要介绍Logstash的入门使用，Elasticsearch默认的安装和配置就已经满足我们要求。
言归正专，现在Elasticsearch已经运行并监听9200端口了，通过简单的设置Logstash就可以使用Elasticsearch作为它的后端。默认的配置对于Logstash和Elasticsearch已经足够，我们忽略一些额外的选项来设置elasticsearch作为output：
bin/logstash -e 'input { stdin { } } output { elasticsearch { host => localhost } }'随意的输入一些字符，Logstash会像之前一样处理日志(不过这次我们将不会看到任何的输出，因为我们没有设置stdout作为output选项)
you know, for logs我们可以使用curl命令发送请求来查看ES是否接收到了数据：
curl 'http://localhost:9200/_search?pretty'返回内容如下：
{ "took" : 2, "timed_out" : false, "_shards" : { "total" : 5, "successful" : 5, "failed" : 0 }, "hits" : { "total" : 1, "max_score" : 1.0, "hits" : [ { "_index" : "logstash-2013.11.21", "_type" : "logs", "_id" : "2ijaoKqARqGvbMgP3BspJA", "_score" : 1.0, "_source" : {"message":"you know, for logs","@timestamp":"2013-11-21T18:45:09.862Z","@version":"1","host":"my-laptop"} } ] }}恭喜，至此你已经成功利用Elasticsearch和Logstash来收集日志数据了。
Elasticsearch 插件(题外话)
这里介绍另外一个对于查询你的Logstash数据(Elasticsearch中数据 )非常有用的工具叫Elasticsearch-kopf插件。更多的信息请见 Elasticsearch插件 。安装elasticsearch-kopf，只要在你安装Elasticsearch的目录中执行以下命令即可：
bin/plugin -install lmenezes/elasticsearch-kopf接下来访问 http://localhost:9200/_plugin/kopf 来浏览保存在Elasticsearch中的数据，设置及映射！
多重输出
作为一个简单的例子来设置多重输出，让我们同时设置stdout和elasticsearch作为output来重新运行一下Logstash，如下：
bin/logstash -e 'input { stdin { } } output { elasticsearch { host => localhost } stdout { } }'当我们输入了一些词组之后，这些输入的内容回回显到我们的终端，同时还会保存到Elasticsearch！(可以使用curl和kopf插件来验证)。
默认配置 - 按照每日日期建立索引
你将发现Logstash可以足够灵巧的在Elasticsearch上建立索引... 每天会按照默认格式是logstash-YYYY.MM.DD来建立索引。在午夜(GMT)，Logstash自动按照时间戳更新索引。我们可以根据追溯多长时间的数据作为依据来制定保持多少数据，当然你也可以把比较老的数据迁移到其他的地方(重新索引)来方便查询，此外如果仅仅是简单的删除一段时间数据我们可以使用 Elasticsearch Curator 。
接下来
接下来我们开始了解更多高级的配置项。在下面的章节，我们着重讨论logstash一些核心的特性，以及如何和logstash引擎交互的。
事件的生命周期
Inputs,Outputs,Codecs,Filters构成了Logstash的核心配置项。Logstash通过建立一条事件处理的管道，从你的日志提取出数据保存到Elasticsearch中，为高效的查询数据提供基础。为了让你快速的了解Logstash提供的多种选项，让我们先讨论一下最常用的一些配置。更多的信息，请参考 Logstash事件管道 。
Inputs
input 及输入是指日志数据传输到Logstash中。其中常见的配置如下：
file：从文件系统中读取一个文件，很像UNIX命令 "tail -0a"
syslog：监听514端口，按照RFC3164标准解析日志数据
redis：从redis服务器读取数据，支持channel(发布订阅)和list模式。redis一般 在Logstash消费集群中 作为"broker"角色，保存events队列共Logstash消费。
lumberjack：使用lumberjack协议来接收数据，目前已经改为 logstash-forwarder。
Filters
Fillters 在Logstash处理链中担任中间处理组件。他们经常被组合起来实现一些特定的行为来，处理匹配特定规则的事件流。常见的filters如下：
grok：解析无规则的文字并转化为有结构的格式。Grok 是目前最好的方式来将无结构的数据转换为有结构可查询的数据。有120多种匹配规则，会有一种满足你的需要。
mutate：mutate filter 允许改变输入的文档，你可以从命名，删除，移动或者修改字段在处理事件的过程中。
drop：丢弃一部分events不进行处理，例如：debug events。
clone：拷贝 event，这个过程中也可以添加或移除字段。
geoip：添加地理信息(为前台kibana图形化展示使用)
Outputs
outputs是logstash处理管道的最末端组件。一个event可以在处理过程中经过多重输出，但是一旦所有的outputs都执行结束，这个event也就完成生命周期。一些常用的outputs包括：
elasticsearch：如果你计划将高效的保存数据，并且能够方便和简单的进行查询...Elasticsearch是一个好的方式。是的，此处有做广告的嫌疑,呵呵。
file：将event数据保存到文件中。
graphite：将event数据发送到图形化组件中，一个很流行的开源存储图形化展示的组件。 http://graphite.wikidot.com/ 。
statsd：statsd是一个统计服务，比如技术和时间统计，通过udp通讯，聚合一个或者多个后台服务，如果你已经开始使用statsd，该选项对你应该很有用。
Codecs
codecs 是基于数据流的过滤器，它可以作为input，output的一部分配置。Codecs可以帮助你轻松的分割发送过来已经被序列化的数据。流行的codecs包括 json,msgpack,plain(text)。
json：使用json格式对数据进行编码/解码
multiline：将汇多个事件中数据汇总为一个单一的行。比如：java异常信息和堆栈信息
获取完整的配置信息，请参考 Logstash文档 中 "plugin configuration"部分。
更多有趣Logstash内容
使用配置文件
使用-e参数 在命令行中指定配置是很常用的方式，不过如果需要配置更多设置则需要很长的内容。这种情况，我们首先创建一个简单的配置文件，并且指定logstash使用这个配置文件。如我们创建一个文件名是"logstash-simple.conf"的配置文件并且保存在和Logstash相同的目录中。内容如下：
input { stdin { } }
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
接下来，执行命令：
bin/logstash -f logstash-simple.conf我们看到logstash按照你刚刚创建的配置文件来运行例子，这样更加的方便。注意，我们使用-f参数来从文件获取而代替之前使用-e参数从命令行中获取配置。以上演示非常简单的例子，当然解析来我们继续写一些复杂一些的例子。
过滤器
filters是一个行处理机制将提供的为格式化的数据整理成你需要的数据，让我们看看下面的一个例子，叫grok filter的过滤器。
input { stdin { } }
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
}
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
执行Logstash按照如下参数：
bin/logstash -f logstash-filter.conf现在粘贴下面一行信息到你的终端(当然Logstash就会处理这个标准的输入内容)：
127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] "GET /xampp/status.php HTTP/1.1" 200 3891 "http://cadenza/xampp/navi.php" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0"你将看到类似如下内容的反馈信息：
{
"message" => "127.0.0.1 - - [11/Dec/2013:00:01:45 -0800] \"GET /xampp/status.php HTTP/1.1\" 200 3891 \"http://cadenza/xampp/navi.php\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\"",
"@timestamp" => "2013-12-11T08:01:45.000Z",
"@version" => "1",
"host" => "cadenza",
"clientip" => "127.0.0.1",
"ident" => "-",
"auth" => "-",
"timestamp" => "11/Dec/2013:00:01:45 -0800",
"verb" => "GET",
"request" => "/xampp/status.php",
"httpversion" => "1.1",
"response" => "200",
"bytes" => "3891",
"referrer" => "\"http://cadenza/xampp/navi.php\"",
"agent" => "\"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:25.0) Gecko/20100101 Firefox/25.0\""
}
正像你看到的那样，Logstash(使用了grok过滤器)能够将一行的日志数据(Apache的"combined log"格式)分割设置为不同的数据字段。这一点对于日后解析和查询我们自己的日志数据非常有用。比如：HTTP的返回状态码，IP地址相关等等，非常的容易。很少有匹配规则没有被grok包含，所以如果你正尝试的解析一些常见的日志格式，或许已经有人为了做了这样的工作。如果查看详细匹配规则，参考 logstash grok patterns 。
另外一个过滤器是date filter。这个过滤器来负责解析出来日志中的时间戳并将值赋给timestame字段(不管这个数据是什么时候收集到logstash的)。你也许注意到在这个例子中@timestamp字段是设置成December 11, 2013, 说明logstash在日志产生之后一段时间进行处理的。这个字段在处理日志中回添到数据中的，举例来说... 这个值就是logstash处理event的时间戳。
实用的例子
Apache 日志(从文件获取)
现在，让我们使用一些非常实用的配置... apache2访问日志！我们将从本地读取日志文件，并且通过条件设置处理满足我们需要的event。首先，我们创建一个文件名是logstash-apache.conf的配置文件，内容如下(你可以根据实际情况修改你的文件名和路径)：
input {
file {
path => "/tmp/access_log"
start_position => beginning
}
}
filter {
if [path] =~ "access" {
mutate { replace => { "type" => "apache_access" } }
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
}
output {
elasticsearch {
host => localhost
}
stdout { codec => rubydebug }
}
接下来，我们按照上面的配置创建一个文件(在例子中是"/tmp/access.log"),可以将下面日志信息作为文件内容(也可以用你自己的webserver产生的日志)：
71.141.244.242 - kurt [18/May/2011:01:48:10 -0700] "GET /admin HTTP/1.1" 301 566 "-" "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US; rv:1.9.2.3) Gecko/20100401 Firefox/3.6.3" 134.39.72.245 - - [18/May/2011:12:40:18 -0700] "GET /favicon.ico HTTP/1.1" 200 1189 "-" "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729; InfoPath.2; .NET4.0C; .NET4.0E)" 98.83.179.51 - - [18/May/2011:19:35:08 -0700] "GET /css/main.css HTTP/1.1" 200 1837 "http://www.safesand.com/information.htm" "Mozilla/5.0 (Windows NT 6.0; WOW64; rv:2.0.1) Gecko/20100101 Firefox/4.0.1"现在使用-f参数来执行一下上面的例子：
bin/logstash -f logstash-apache.conf你可以看到apache的日志数据已经导入到ES中了。这里logstash会按照你的配置读取，处理指定的文件，任何后添加到文件的内容也会被捕获处理最后保存到ES中。此外，数据中type的字段值会被替换成"apache_access"(这个功能在配置中已经指定)。
这个配置只是让Logstash监控了apache access_log，但是在实际中往往并不够用可能还需要监控error_log，只要在上面的配置中改变一行既可以实现，如下：
input {
file {
path => "/tmp/*_log"
...
现在你可以看到logstash处理了error日志和access日志。然而，如果你检查了你的数据(也许用elasticsearch-kopf )，你将发现access_log日志被分成不同的字段，但是error_log确没有这样。这是因为我们使用了“grok”filter并仅仅配置匹配combinedapachelog日志格式，这样满足条件的日志就会自动的被分割成不同的字段。我们可以通过控制日志按照它自己的某种格式来解析日志，不是很好的吗？对吧。
此外，你也许还会发现Logstash不会重复处理文件中已经处理过得events。因为Logstash已经记录了文件处理的位置，这样就只处理文件中新加入的行数。漂亮！
条件判断
我们利用上一个例子来介绍一下条件判断的概念。这个概念一般情况下应该被大多数的Logstash用户熟悉掌握。你可以像其他普通的编程语言一样来使用if,else if和else语句。让我们把每个event依赖的日志文件类型都标记出来(access_log,error_log其他以log结尾的日志文件)。
input {
file {
path => "/tmp/*_log"
}
}
filter {
if [path] =~ "access" {
mutate { replace => { type => "apache_access" } }
grok {
match => { "message" => "%{COMBINEDAPACHELOG}" }
}
date {
match => [ "timestamp" , "dd/MMM/yyyy:HH:mm:ss Z" ]
}
} else if [path] =~ "error" {
mutate { replace => { type => "apache_error" } }
} else {
mutate { replace => { type => "random_logs" } }
}
}
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
我想你已经注意到了，我们使用"type"字段来标记每个event,但是我们实际上没有解析"error"和”random"类型的日志... 而实际情况下可能会有很多很多类型的错误日志，如何解析就作为练习留给各位读者吧，你可以依赖已经存在的日志。
Syslog
Ok,现在我们继续了解一个很实用的例子：syslog。Syslog对于Logstash是一个很长用的配置，并且它有很好的表现(协议格式符合RFC3164)。Syslog实际上是UNIX的一个网络日志标准，由客户端发送日志数据到本地文件或者日志服务器。在这个例子中，你根本不用建立syslog实例;我们通过命令行就可以实现一个syslog服务，通过这个例子你将会看到发生什么。
首先，让我们创建一个简单的配置文件来实现logstash+syslog，文件名是 logstash-syslog.conf
input {
tcp {
port => 5000
type => syslog
}
udp {
port => 5000
type => syslog
}
}
filter {
if [type] == "syslog" {
grok {
match => { "message" => "%{SYSLOGTIMESTAMP:syslog_timestamp} %{SYSLOGHOST:syslog_hostname} %{DATA:syslog_program}(?:\[%{POSINT:syslog_pid}\])?: %{GREEDYDATA:syslog_message}" }
add_field => [ "received_at", "%{@timestamp}" ]
add_field => [ "received_from", "%{host}" ]
}
syslog_pri { }
date {
match => [ "syslog_timestamp", "MMM d HH:mm:ss", "MMM dd HH:mm:ss" ]
}
}
}
output {
elasticsearch { host => localhost }
stdout { codec => rubydebug }
}
执行logstash：
bin/logstash -f logstash-syslog.conf通常，需要一个客户端链接到Logstash服务器上的5000端口然后发送日志数据。在这个简单的演示中我们简单的使用telnet链接到logstash服务器发送日志数据(与之前例子中我们在命令行标准输入状态下发送日志数据类似)。首先我们打开一个新的shell窗口，然后输入下面的命令：
telnet localhost 5000你可以复制粘贴下面的样例信息(当然也可以使用其他字符，不过这样可能会被grok filter不能正确的解析)：
Dec 23 12:11:43 louis postfix/smtpd[31499]: connect from unknown[95.75.93.154] Dec 23 14:42:56 louis named[16000]: client 199.48.164.7#64817: query (cache) 'amsterdamboothuren.com/MX/IN' deniedDec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log) Dec 22 18:28:06 louis rsyslogd: [origin software="rsyslogd" swVersion="4.2.0" x-pid="2253" x-info="http://www.rsyslog.com"] rsyslogd was HUPed, type 'lightweight'.之后你可以在你之前运行Logstash的窗口中看到输出结果，信息被处理和解析！
{
"message" => "Dec 23 14:30:01 louis CRON[619]: (www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
"@timestamp" => "2013-12-23T22:30:01.000Z",
"@version" => "1",
"type" => "syslog",
"host" => "0:0:0:0:0:0:0:1:52617",
"syslog_timestamp" => "Dec 23 14:30:01",
"syslog_hostname" => "louis",
"syslog_program" => "CRON",
"syslog_pid" => "619",
"syslog_message" => "(www-data) CMD (php /usr/share/cacti/site/poller.php >/dev/null 2>/var/log/cacti/poller-error.log)",
"received_at" => "2013-12-23 22:49:22 UTC",
"received_from" => "0:0:0:0:0:0:0:1:52617",
"syslog_severity_code" => 5,
"syslog_facility_code" => 1,
"syslog_facility" => "user-level",
"syslog_severity" => "notice"
}
看到这里你已经成为一个合格的Logstash用户了。你将可以轻松的配置，运行Logstash，还可以发送event给Logstash，但是这个过程随着使用还会有很多值得深挖的地方。
参考：Logstash官方网站文档，多数细节来源于官方文档描述，小编加以翻译与个人理解整理而成。
推荐阅读
·end·
—写文不易，你的转发就是对我最大的支持—
我们一起愉快的玩耍吧
目前40000+人已关注加入我们
关注公众号点击菜单“微信群” 入群与小伙伴一起交流吧！